# coding=utf-8
import json
import os
import sys
import boto3
from smart_open import open
import subprocess

sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))

job_list_path = "s3://streaming/flink/conf/job-list-"
s3 = boto3.client("s3")
sns = boto3.client("sns")

transport_params = {'client': s3}


# 读取任务启动参数文件
def read_job_conf(env):
    path = f"{job_list_path}{env}.json"
    print(f"read job list conf {path}")
    with open(path, 'r', transport_params=transport_params) as file:
        conf = json.load(file)
        names = []
        for name, p in conf.items():
            if "enable" in p and str(p['enable']).lower() == "false":
                names.append(name)
        for name in names:
            del conf[name]
    if not conf:
        raise Exception(f" job list file {path} is not available")
    return conf


def get_env():
    instance_info = subprocess.check_output("cat /mnt/var/lib/info/job-flow.json", shell=True)
    cluster_id = json.loads(instance_info)['jobFlowId']
    cluster_info = subprocess.check_output(f"aws emr describe-cluster --cluster-id {cluster_id}", shell=True)
    cluster_name = json.loads(cluster_info)['Cluster']['Name']
    if cluster_name == "streaming-prod":
        env = "prod"
    elif cluster_name == "streaming-test":
        env = "test"
    else:
        raise Exception(f" emr cluster name error,need streaming-prod or streaming-test")
    return env


def monitor(job_name):
    prefix = " yarn application -list |grep " + job_name + " | awk '{print $1}' "
    import subprocess
    print(prefix)
    rs = subprocess.check_output(prefix, shell=True).decode().split("\n")
    if len(rs) >= 2:
        print(" job ", job_name, "is running")
        return ""
    else:
        return job_name


def send_msg(job_name):
    print("job ", job_name, "not running")
    msg = {job_name: "任务未运行"}


def run():
    # 获取环境 test或者prod
    env = get_env()

    # 读取需要运行的任务列表
    job_list = read_job_conf(env)

    # 轮询job_list，查看每一个job是否正在运行
    not_running_list = []
    for job_name, pa in job_list.items():
        r = monitor(job_name)
        if r != "":
            not_running_list.append(r)

    # 发送未运行的任务监控消息
    j = {}
    if env == "prod":
        env_ch = "正式环境-"
    elif env == "test":
        env_ch = "测试环境-"
    else:
        env_ch = "未知环境-"
    for one in not_running_list:
        j[env_ch + one] = "任务未运行"
    if len(j) > 0:
        response = sns.publish(
            TopicArn='arn:aws-cn:sns:cn-northwest-1:453200521012:Monthly_buisiness_alarm',
            Message=json.dumps(j)
        )
        print(response)


if __name__ == '__main__':
    run()
